package com.safety.wc;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @Author: QingW
 * @Date: 2021/9/15
 */
public class StreamWordCount {
    public static void main(String[] args) throws Exception {

//        流处理  人员定位的数据 源源不断的会传送过来
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        env.setParallelism(1);
//        String inputPath = "E:\\flink\\flink-learning\\src\\main\\resources\\hello.txt";
//        DataStreamSource<String> inputDataStream = env.readTextFile(inputPath);

        ParameterTool parameterTool = ParameterTool.fromArgs(args);
        String host = parameterTool.get("host");
        int port = parameterTool.getInt("port");
        DataStreamSource<String> inputDataStream = env.socketTextStream(host, port);

//        DataStream<Tuple2<String, Integer>> resultStream = inputDataStream.flatMap(new WordCount.MyFlatMapper()).keyBy(0).sum(1).slotSharingGroup("red").shuffle();
        DataStream<Tuple2<String, Integer>> resultStream = inputDataStream.flatMap(new WordCount.MyFlatMapper()).keyBy(0).sum(1).slotSharingGroup("red").disableChaining();
        resultStream.print();
//        执行任务
        env.execute();
//        执行的时候 并行执行的线程
    }

}
